package lyw.app.dwd.db;

import lyw.app.BaseSqlApp;
import lyw.common.Constant;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import lyw.util.SQLUtil;

public class DwdTradeUserOrderDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeUserOrderDetail().init(3003, 2, "DwdTradeUserOrderDetail");
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        // 1. 读取ods_db数据1
        readOdsDb(tEnv, "DwdTradeUserOrderDetail");

        // 2. 过滤订单明细表
        Table orderDetail = tEnv.sqlQuery(
                "select " +
                        " data['id'] id, " +
                        " data['course_id'] course_id, " +
                        " data['course_name'] course_name, " +
                        " data['order_id'] order_id, " +
                        " data['user_id'] user_id, " +
                        " data['origin_amount'] origin_amount," +
                        " data['coupon_amount'] coupon_amount," +
                        " data['final_amount'] final_amount," +
                        " `ts` " +
                        " from ods_db " +
                        " where `database`='edu' " +
                        " and `table`='order_detail' " +
                        " and `type`='insert'"
        );
        tEnv.createTemporaryView("order_detail",orderDetail);


        // 3. 过滤订单
        Table orderInfo = tEnv.sqlQuery(
                "select " +
                        " data['id'] id, " +
                        " data['user_id'] user_id, " +
                        " data['origin_amount'] origin_amount," +
                        " data['coupon_amount'] coupon_amount," +
                        " data['final_amount'] final_amount," +
                        " data['order_status'] order_status," +
                        " `ts` " +
                        " from ods_db " +
                        " where `database`='edu' " +
                        " and `table`='order_info' " +
                        " and `type`='insert'"
        );
        tEnv.createTemporaryView("order_info",orderInfo);

        Table result = tEnv.sqlQuery(
                "select " +
                        " od.user_id, " +
                        " od.order_id, " +
                        " oi.order_status, " +
                        " od.course_id," +
                        " od.course_name, " +
                        " od.origin_amount, " +
                        " od.coupon_amount, " +
                        " od.final_amount," +
                        " od.ts" +
                        " from order_detail od " +
                        " join order_info oi " +
                        " on od.order_id = oi.id "
        );

        // 3. 写出到kafka中
        tEnv.executeSql(
                "create table dwd_trade_user_order_detail ( " +
                        " user_id string, " +
                        " order_id string, " +
                        " order_status string, " +
                        " course_id string, " +
                        " course_name string, " +
                        " origin_amount string, " +
                        " coupon_amount string, " +
                        " final_amount string, " +
                        " ts bigint " +
                        " ) " + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_USER_ORDER_DETAIL)
        );

        result.executeInsert("dwd_trade_user_order_detail");
    }
}
